package com.atguigu.daijia.common.config;

import com.alibaba.fastjson.JSON;
import com.atguigu.daijia.common.entity.GuiguCorrelationData;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @Author XiangLiang
 * @Since 2024/12/26 17:44
 * @Description: RabbitInitConfigApplicationListener类
 */
@Slf4j
@Component
public class RabbitInitConfigApplicationListener implements ApplicationListener<ApplicationReadyEvent> {
    @Resource
    private RabbitTemplate rabbitTemplate;

    @Resource
    private RedisTemplate<Object, Object> redisTemplate;

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        this.setupCallbacks();
    }

    private void setupCallbacks() {
        /*
         * 只确认消息是否正确到达 Exchange 中,成功与否都会回调
         *
         * @param correlation 相关数据  非消息本身业务数据
         * @param ack         应答结果
         * @param reason      如果发送消息到交换器失败，错误原因
         */
        this.rabbitTemplate.setConfirmCallback((correlation, ack, reason) -> {
            if (ack) {
                // 消息到交换器成功
                log.info("消息发送到Exchange成功：{}", correlation);
            } else {
                // 消息到交换器失败
                log.error("消息发送到Exchange失败：{}", reason);
                // 执行消息重发
                this.retrySendMsg(correlation);
            }
        });

        /*
         * 消息没有正确到达队列时触发回调，如果正确到达队列不执行
         */
        this.rabbitTemplate.setReturnsCallback(returned -> {
            log.error("Returned: {}\n replyCode: {}\n replyText: {}\n exchange/r k: {}/{}",
                    returned.getMessage(),
                    returned.getReplyCode(),
                    returned.getReplyText(),
                    returned.getExchange(),
                    returned.getRoutingKey());

            // 当路由队列失败 也需要重发
            // 1.构建相关数据对象
            String redisKey = returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation");
            String correlationDataStr = (String) redisTemplate.opsForValue().get(redisKey);
            GuiguCorrelationData correlationData = JSON.parseObject(correlationDataStr, GuiguCorrelationData.class);
            // 方式一:如果不考虑延迟消息重发 直接返回
            if (correlationData == null || correlationData.isDelay()) {
                return;
            }
            // 2.调用消息重发方法
            this.retrySendMsg(correlationData);
        });
    }

    /**
     * 消息重新发送
     *
     * @param correlation 消息相关数据
     */
    private void retrySendMsg(CorrelationData correlation) {
        // 获取相关数据
        GuiguCorrelationData correlationData = (GuiguCorrelationData) correlation;

        // 获取redis中存放重试次数
        // 先重发，在写会到redis中次数
        int retryCount = correlationData.getRetryCount();
        if (retryCount >= 3) {
            // 超过最大重试次数
            log.error("生产者超过最大重试次数，将失败的消息存入数据库用人工处理；给管理员发送邮件；给管理员发送短信；");
            return;
        }
        // 重发消息
        /*rabbitTemplate.convertAndSend(correlationData.getExchange(),
                correlationData.getRoutingKey(),
                correlationData.getMessage(),
                correlationData);*/

        // 重发次数+1
        retryCount += 1;
        correlationData.setRetryCount(retryCount);
        redisTemplate.opsForValue().set(correlationData.getId(), JSON.toJSONString(correlationData), 10, TimeUnit.MINUTES);
        log.info("进行消息重发！");

        // 重发消息
        // 方式二：如果是延迟消息，依然需要设置消息延迟时间
        if (correlationData.isDelay()) {
            // 延迟消息
            rabbitTemplate.convertAndSend(correlationData.getExchange(),
                    correlationData.getRoutingKey(),
                    correlationData.getMessage(),
                    message -> {
                        message.getMessageProperties().setDelay(correlationData.getDelayTime() * 1000);
                        return message;
                    }, correlationData);
        } else {
            // 普通消息
            rabbitTemplate.convertAndSend(correlationData.getExchange(),
                    correlationData.getRoutingKey(),
                    correlationData.getMessage(),
                    correlationData);
        }
    }
}
